
package org.example.datastream.test;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema;
import org.example.util.Utils;
import org.example.datastream.test.custom.Enterprise;
import org.example.datastream.test.custom.EnterpriseJdbcSink;
import org.example.datastream.test.custom.RecordLog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.util.Properties;

/**
 * 增量数据同步
 * 数据源为kafka
 *
 * [
 *     {
 *         "head": {
 *             "binlog_pos": 90546640,
 *             "time": 1618973029000,
 *             "type": "UPDATE",
 *             "binlog_file": "330617124-bin.000512",
 *             "db": "db_enterprise_3",
 *             "table": "t_enterprise_15"
 *         },
 *         "before": {
 *             "actual_capi": "",
 *             "reg_no": "150902600208895",
 *             "logo_url": "",
 *             "collegues_num": "0人",
 *             "source": "1",
 *             "type": 0,
 *             "province_code": "150000",
 *             "gd_longitude": 113.12614,
 *             "new_status_code": "1",
 *             "province": "NMG",
 *             "check_date": "2016-03-11",
 *             "row_update_time": 1593776958000,
 *             "belong_org": "乌兰察布市工商行政管理局集宁分局",
 *             "term_end": "",
 *             "id": 725587,
 *             "type_desc": "",
 *             "org_no": "",
 *             "credit_area_code": "",
 *             "econ_kind": "个体工商户",
 *             "longitude": 113.13214765235,
 *             "created_time": 1485712014560,
 *             "group_name": "",
 *             "group_abbreviation": "",
 *             "oper_name": "孙小波",
 *             "type_new": "01",
 *             "title_code": "999A",
 *             "belong_org_code": "150902",
 *             "last_update_time": 1506662372872,
 *             "econ_type": "",
 *             "category_new": "0115602",
 *             "district_code": "150902",
 *             "logout_reason": "",
 *             "name": "乌兰察布市集宁区家常水煎包子",
 *             "revoke_date": "",
 *             "format_name": "乌兰察布市集宁区家常水煎包子",
 *             "revoked_certificates": "",
 *             "status": "存续（在营、开业、在册）",
 *             "end_date": "",
 *             "eid": "eb1210ef-376d-40ab-aed9-c1d3487bd4f3",
 *             "status_code": "",
 *             "revoke_reason": "",
 *             "latitude": 41.040761266407,
 *             "description": "",
 *             "gd_latitude": 41.03268,
 *             "title": "",
 *             "logout_date": "",
 *             "tax_no": "",
 *             "scope": "许可经营项目：中餐类制售、不含凉菜裱花蛋糕服务。 一般经营项目：无",
 *             "department": "",
 *             "start_date": "2016-03-11",
 *             "econ_kind_code": "9600",
 *             "address": "",
 *             "obj_id": "588e2a7b8587822df62f1341",
 *             "oper_name_id": "033c54edf1ea336fe611231f2d888607",
 *             "org_type": "",
 *             "currency_unit": "CNY",
 *             "url": "",
 *             "credit_no": "",
 *             "term_start": "",
 *             "oper_type": "P",
 *             "regist_capi_new": 1.000000,
 *             "category": "P",
 *             "regist_capi": "1万元人民币"
 *         },
 *         "pk": "eid",
 *         "after": {
 *             "actual_capi": "",
 *             "reg_no": "150902600208895",
 *             "logo_url": "",
 *             "collegues_num": "",
 *             "source": "1",
 *             "type": 0,
 *             "province_code": "150000",
 *             "gd_longitude": 113.12614,
 *             "new_status_code": "1",
 *             "province": "NMG",
 *             "check_date": "2016-03-11",
 *             "row_update_time": 1619001829000,
 *             "belong_org": "乌兰察布市集宁区市场监督管理局",
 *             "term_end": "",
 *             "id": 725587,
 *             "type_desc": "",
 *             "org_no": "MA0PJMEL4",
 *             "credit_area_code": "150902",
 *             "econ_kind": "个体工商户",
 *             "longitude": 113.13214765235,
 *             "created_time": 1485712014560,
 *             "group_name": "",
 *             "group_abbreviation": "",
 *             "oper_name": "孙小波",
 *             "type_new": "01",
 *             "title_code": "999A",
 *             "belong_org_code": "150902",
 *             "last_update_time": 1506662372872,
 *             "econ_type": "",
 *             "category_new": "0115602",
 *             "district_code": "150902",
 *             "logout_reason": "",
 *             "name": "乌兰察布市集宁区家常水煎包子",
 *             "revoke_date": "",
 *             "format_name": "乌兰察布市集宁区家常水煎包子",
 *             "revoked_certificates": "",
 *             "status": "存续（在营、开业、在册）",
 *             "end_date": "",
 *             "eid": "eb1210ef-376d-40ab-aed9-c1d3487bd4f3",
 *             "status_code": "",
 *             "revoke_reason": "",
 *             "latitude": 41.040761266407,
 *             "description": "",
 *             "gd_latitude": 41.03268,
 *             "title": "",
 *             "logout_date": "",
 *             "tax_no": "",
 *             "scope": "许可经营项目：中餐类制售、不含凉菜裱花蛋糕服务。 一般经营项目：无",
 *             "department": "",
 *             "start_date": "2016-03-11",
 *             "econ_kind_code": "9600",
 *             "address": "",
 *             "obj_id": "588e2a7b8587822df62f1341",
 *             "oper_name_id": "033c54edf1ea336fe611231f2d888607",
 *             "org_type": "",
 *             "currency_unit": "CNY",
 *             "url": "",
 *             "credit_no": "92150902MA0PJMEL49",
 *             "term_start": "",
 *             "oper_type": "P",
 *             "regist_capi_new": 1.000000,
 *             "category": "P",
 *             "regist_capi": "1万元人民币"
 *         }
 *     }
 * ]
 */
public class Job3 {
	private static final Logger LOGGER = LoggerFactory.getLogger(Job3.class);

	public static void main(String[] args) throws Exception {
		// set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		env.enableCheckpointing(5000);


		Properties properties = new Properties();
		properties.setProperty("bootstrap.servers", "localhost:9092");
		properties.setProperty("group.id", "df-flink-test");
		String topic = "sync_v2_db_enterprise_t_enterprise";
		FlinkKafkaConsumer<ObjectNode> consumer = new FlinkKafkaConsumer<>(topic,
				new JSONKeyValueDeserializationSchema(false), properties);

		env.addSource(consumer)
				.name("t_enterprise_x")
				.map(new MapFunction<ObjectNode, RecordLog<Enterprise>>() {
					@Override
					public RecordLog<Enterprise> map(ObjectNode json) throws Exception {
						JsonNode value = json.get("value");

						if (value.isArray()) {
							JsonNode n = value.get(0);

							JsonNode head = n.get("head");
							JsonNode before = n.get("before");
							JsonNode after = n.get("after");

							String type = head.get("type").asText();
							String tableName = head.get("table").asText();
							String table = null;
							if (tableName.lastIndexOf("_") == tableName.length()-1-1
									|| tableName.lastIndexOf("_") == tableName.length()-1-2) {

								table = tableName.substring(0, tableName.lastIndexOf("_"));
							}

							String pk = n.get("pk").asText();
							String pkv = null;
							Enterprise ent = new Enterprise();
							if ("INSERT".equals(type) || "UPDATE".equals(type)) {
								pkv = after.get(pk).asText();

								ent.setEid(after.get("eid").asText());
								ent.setEnterpriseName(after.get("name").asText());
								ent.setLogoUrl(after.get("logo_url").asText());
								ent.setOperName(after.get("oper_name").asText());
								ent.setBusinessUrl(after.get("url").asText());

								ent.setDepartment(after.get("department").asText());
								ent.setType(after.get("type").asInt());
								ent.setStatusCode(after.get("new_status_code").asInt());
								ent.setCreateTime(Timestamp.valueOf(Utils.timestamp2Datetime(after.get("created_time").asLong())));
								ent.setRowUpdateTime(Timestamp.valueOf(Utils.timestamp2Datetime(after.get("row_update_time").asLong())));

								// 其他字段省略 ...
							} else {
								pkv = before.get(pk).asText();

								ent.setEid(before.get("eid").asText());
								ent.setEnterpriseName(before.get("name").asText());
								ent.setLogoUrl(before.get("logo_url").asText());
								ent.setOperName(before.get("oper_name").asText());
								ent.setBusinessUrl(before.get("url").asText());

								ent.setDepartment(before.get("department").asText());
								ent.setType(before.get("type").asInt());
								ent.setStatusCode(before.get("new_status_code").asInt());
								ent.setCreateTime(Timestamp.valueOf(Utils.timestamp2Datetime(before.get("created_time").asLong())));
								ent.setRowUpdateTime(Timestamp.valueOf(Utils.timestamp2Datetime(before.get("row_update_time").asLong())));

								// 其他字段省略 ...
							}

							return new RecordLog<>(table,type, pk, pkv, ent);
						}
						LOGGER.error("kafka message is not json array");
						return null;
					}
				})
				.addSink(new EnterpriseJdbcSink(
						"mysql",
						"com.mysql.jdbc.Driver",
						"localhost",
						"3306",
						"db_test",
						"flinksql",
						"w0c53w0c53"
				))
				.name("t_enterprise");


		// execute program
		env.execute("Flink Streaming Job test");
	}
}
